package com.qit.prototype.flink.performance;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import com.qit.common.stream.ParallelSequenceProcessor;
import com.qit.prototype.flink.entity.MarketData;
import com.qit.prototype.flink.entity.Tick;
import com.qit.prototype.flink.processor.IMarketDataConvertProcessor;
import com.qit.prototype.flink.processor.MarketDataConvertJob;
import com.qit.prototype.flink.processor.MarketDataConvertProcessor;

/**
 * Compare serial and parallel sequence processing.
 * 
 * @author rollgan
 *
 */
public class TestCase9 {
	
	public static BlockingQueue<MarketData> outputMarketData = new LinkedBlockingQueue<MarketData>();
	public static List<Tick> tickSource = new ArrayList<Tick>();
	
	private StreamExecutionEnvironment env;
	
	private int mode = 4;
	private long batchCount = 100000;
	
	private long beginTime;
	
	public TestCase9() throws Exception {
		
		// Source.
		// Local entities.
		System.out.println("create source.");
		tickSource = new ArrayList<Tick>();
		
		for (int i = 0; i < batchCount; i++) {
			Tick tick = new Tick();
			tick.setId(i + 1L);
			tick.setPrice(1.0);
			tick.setQty(i + 1.0);
			tickSource.add(tick);
		}
		
		// Create a thread to write market data in output.
		System.out.println("create comsumer.");
		Thread t = new Thread(new Runnable() {

			@Override
			public void run() {
				while (true) {
					try {
						MarketData md = outputMarketData.take();
//						System.out.println("output: md.fieldValues=" + md.getFieldValues());
						if (md.getId() == 1) {
							beginTime = System.currentTimeMillis();
						}
						else if (md.getId() == batchCount) {
							long endTime = System.currentTimeMillis();
							long usedTime = endTime - beginTime;
							System.out.println("usedTime=" + usedTime);
							System.out.println("output: md.fieldValues=" + md.getFieldValues());
						}
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		});
		
		t.start();
		
		// Market used time.
		if (mode == 1) {
			doInSerial();
		}
		else if (mode == 2) {
			initParallelEnv();
			doInParallel();
		}
		else if (mode == 3) {
			doInParallelCustom();
		}
		else if (mode == 4) {
			doInParallelSequenceProcessor();
		}
	}
	
	private void doInSerial() {
		for (int i = 0; i < tickSource.size(); i++) {
			Tick tick = tickSource.get(i);
			MarketData md = convertToMarketData(tick);
			try {
				outputMarketData.put(md);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	private void doInParallel() {
		try {
			env.execute();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	private void doInParallelCustom() {
		int workerCount = 8;
		List<BlockingQueue<Tick>> sourceQueues = new ArrayList<BlockingQueue<Tick>>();
		List<BlockingQueue<MarketData>> targetQueues = new ArrayList<BlockingQueue<MarketData>>();
		for (int i = 0; i < workerCount; i++) {
			BlockingQueue<Tick> sourceQueue = new LinkedBlockingQueue<Tick>();
			sourceQueues.add(sourceQueue);
			BlockingQueue<MarketData> targetQueue = new LinkedBlockingQueue<MarketData>();
			targetQueues.add(targetQueue);
			Thread t = new Thread(new ParallelWorker(i, sourceQueue, targetQueue));
			t.start();
		}
		
		Thread collector = new Thread(new Runnable() {
			private int currentIndex = 0;

			@Override
			public void run() {
				while (true) {
					try {
						BlockingQueue<MarketData> currentTargetQueue = targetQueues.get(currentIndex);
						MarketData md = currentTargetQueue.take();
						outputMarketData.put(md);
						currentIndex++;
						if (currentIndex >= workerCount) {
							currentIndex = 0;
						}
					}
					catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		});
		collector.start();
		
		Thread dispatcher = new Thread(new Runnable() {
			private int currentIndex;
			
			@Override
			public void run() {
				try {
					for (int i = 0; i < tickSource.size(); i++) {
						Tick tick = tickSource.get(i);
						BlockingQueue<Tick> currentSourceQueue = sourceQueues.get(currentIndex);
						currentSourceQueue.put(tick);
						currentIndex++;
						if (currentIndex >= workerCount) {
							currentIndex = 0;
						}
					}
				}
				catch (Exception e) {
					e.printStackTrace();
				}
			}
		});
		dispatcher.start();
		
		
	}
	
	private void doInParallelSequenceProcessor() {
		IMarketDataConvertProcessor processor = new MarketDataConvertProcessor(8, new MarketDataConvertJob());
		
		/*
		ParallelSequenceProcessor<Tick, MarketData> processor = 
				new ParallelSequenceProcessor<Tick, MarketData>(8, new ParallelSequenceJob<Tick, MarketData>() {
					@Override
					public MarketData execute(Tick source) {
						MarketData target = convertToMarketData(source);
						return target;
					}
				});
		*/
		
		Thread t = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						MarketData target = processor.take();
						outputMarketData.put(target);
					}
					catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		});
		t.start();
		
		try {
			for (int i = 0; i < tickSource.size(); i++) {
				Tick source = tickSource.get(i);
				processor.put(source);
			}
		}
		catch (Exception e) {
			e.printStackTrace();
		}
		
	}
	
	private void initParallelEnv() {
		// Create env.
		env = StreamExecutionEnvironment.getExecutionEnvironment();
		// Env default time characteristic is processing time.
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
		
		DataStream<Tick> dsTick = env.fromCollection(tickSource);
		DataStream<MarketData> dsMarketData = dsTick.map(new MapFunction<Tick, MarketData>() {
			private static final long serialVersionUID = -6284057346699940235L;

			@Override
			public MarketData map(Tick tick) throws Exception {
//				System.out.println("map");
				MarketData md = convertToMarketData(tick);
//				MarketData md = new MarketData(tick.getSymbol(), MarketData.FIELD_GROUP_TRADE);
//				md.setId(tick.getId());
//				md.getFieldValues().put("price", String.valueOf(tick.getPrice()));
//				md.getFieldValues().put("qty", String.valueOf(tick.getQty()));
				return md;
			}
			
		});
		
		dsMarketData = dsMarketData.assignTimestampsAndWatermarks(
//				WatermarkStrategy.<MarketData>forMonotonousTimestamps()
				WatermarkStrategy.<MarketData>forBoundedOutOfOrderness(Duration.ofMillis(10))
				.withTimestampAssigner(new SerializableTimestampAssigner<MarketData>() {
					@Override
					public long extractTimestamp(MarketData element, long recordTimestamp) {
						return element.getId();
					}
				}));
		
		Pattern<MarketData, ?> pattern = Pattern.<MarketData>begin("start")
				.where(new SimpleCondition<MarketData>() {

					private static final long serialVersionUID = -7850969134419767514L;
		
					@Override
					public boolean filter(MarketData md) throws Exception {
						return true;
					}
			
				});
		
		
		PatternStream<MarketData> patternStream = CEP.pattern(dsMarketData, pattern);
		
		patternStream.process(new PatternProcessFunction<MarketData, MarketData>() {

			private static final long serialVersionUID = 1503737342639825382L;

			@Override
			public void processMatch(
					Map<String, List<MarketData>> arg0,
					org.apache.flink.cep.functions.PatternProcessFunction.Context arg1,
					Collector<MarketData> arg2) throws Exception {
				outputMarketData.put(arg0.get("start").get(0));
			}
		});
	}
	
	private static MarketData convertToMarketData(Tick tick) {

		int i = 0;
		while (i < 5000) {
			Random r = new Random(10000);
			r.nextInt();
			i++;
		}
		
		MarketData md = new MarketData(tick.getSymbol(), MarketData.FIELD_GROUP_TRADE);
		md.setId(tick.getId());
		md.getFieldValues().put("price", String.valueOf(tick.getPrice()));
		md.getFieldValues().put("qty", String.valueOf(tick.getQty()));
		return md;
	}
	
	private class ParallelWorker implements Runnable {
		
		private int index;
		private BlockingQueue<Tick> sourceQueue;
		private BlockingQueue<MarketData> targetQueue;
		
		public ParallelWorker(
				int index, 
				BlockingQueue<Tick> sourceQueue, 
				BlockingQueue<MarketData> targetQueue) {
			this.index = index;
			this.sourceQueue = sourceQueue;
			this.targetQueue = targetQueue;
		}

		@Override
		public void run() {
			while (true) {
				try {
					Tick tick = sourceQueue.take();
					MarketData md = convertToMarketData(tick);
					targetQueue.put(md);
				}
				catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}

}
